package com.hlz.flink.chapter05;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 聚合操作
 *
 * @author Hongliang Zhu
 * @create 2023-02-16 23:10
 */
public class TransTupleAggreationTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<Tuple2<String, Integer>> stream =
                env.fromElements(Tuple2.of("a", 1), Tuple2.of("a", 3), Tuple2.of("b", 3), Tuple2.of("b", 4));

        KeyedStream<Tuple2<String, Integer>, String> tuple2StringKeyedStream = stream.keyBy(t -> t.f0);

        /**
         * keyStream:1> (b,3)
         * keyStream:3> (a,1)
         * keyStream:1> (b,4)
         * keyStream:3> (a,3)
         *
         * 同一个key的数据都在同一个slot中执行
         */
//        tuple2StringKeyedStream.print("keyStream");


        /**
         * sum:1> (b,3)
         * sum:3> (a,1)
         * sum:1> (b,7)
         * sum:3> (a,4)
         */
//        tuple2StringKeyedStream.sum(1).print("sum");


        /**
         * sum-f1:3> (a,1)
         * sum-f1:1> (b,3)
         * sum-f1:3> (a,4)
         * sum-f1:1> (b,7)
         */
//        tuple2StringKeyedStream.sum("f1").print("sum-f1");

        /**
         * max:1> (b,3)
         * max:3> (a,1)
         * max:1> (b,4)
         * max:3> (a,3)
         */
//        tuple2StringKeyedStream.max(1).print("max");


        /**
         * min:3> (a,1)
         * min:1> (b,3)
         * min:3> (a,1)
         * min:1> (b,3)
         */
//        tuple2StringKeyedStream.min(1).print("min");


        /**
         * minBy:1> (b,3)
         * minBy:3> (a,1)
         * minBy:1> (b,3)
         * minBy:3> (a,1)
         */
        tuple2StringKeyedStream.minBy(1).print("minBy");

        env.execute();
    }
}
